/*
 * Copyright (c) 2020 - present, Inspur Genersoft Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.iec.edp.caf.rpc.api.channel;

import io.iec.edp.caf.commons.core.SerializerFactory;
import io.iec.edp.caf.commons.core.enums.SerializeType;
import io.iec.edp.caf.commons.utils.SpringBeanUtils;
import io.iec.edp.caf.rpc.api.annotation.RpcClientInterceptor;
import io.iec.edp.caf.rpc.api.annotation.RpcServerInterceptor;
import io.iec.edp.caf.rpc.api.filter.RpcClientFilter;
import io.iec.edp.caf.rpc.api.filter.RpcServerFilter;
import io.iec.edp.caf.rpc.api.support.ConstanceVarible;
import io.iec.edp.caf.rpc.api.support.RpcFiltersContainer;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;

import java.net.URL;
import java.util.*;

/***
 * this class is RpcChanel default implement
 * this class register RpcFilters
 * any RpcChanel implement should extend this class
 * and override all abstract method
 */
@Slf4j
public abstract class RpcAbstractChannel implements RpcChannel {

    public RpcAbstractChannel(){
        collectFilters();
    }

    /**
     * any custom filter is extend for platform common
     * so any custom filters(interceptor) should in platform/common/ext
     *
     * */
    protected void collectFilters(){

        if(RpcFiltersContainer.getClientFilter().size()==0 && RpcFiltersContainer.getServerFilter().size()==0){
            ApplicationContext context = SpringBeanUtils.getApplicationContext();
            Map<String,Object> rpcClientFilter = context.getBeansWithAnnotation(RpcClientInterceptor.class);
            Map<String,Object> rpcServerFilter = context.getBeansWithAnnotation(RpcServerInterceptor.class);

            rpcClientFilter.forEach((beanName,bean)->{
                String mode = bean.getClass().getAnnotation(RpcClientInterceptor.class).mode();
                if(mode==null||"".equals(mode)) mode = ConstanceVarible.RPC_FILTER_MODE_DEFAULT;
                if(RpcFiltersContainer.getClientFilter().containsKey(mode)){
                    RpcFiltersContainer.getClientFilter().get(mode).add((RpcClientFilter)bean);
                }else{
                    List<RpcClientFilter> map = new ArrayList<>();
                    map.add((RpcClientFilter)bean);
                    RpcFiltersContainer.getClientFilter().put(mode,map);
                }
            });

            rpcServerFilter.forEach((beanName,bean)->{
                String mode = bean.getClass().getAnnotation(RpcServerInterceptor.class).mode();
                if(mode==null||"".equals(mode)) mode = ConstanceVarible.RPC_FILTER_MODE_DEFAULT;
                if(RpcFiltersContainer.getServerFilter().containsKey(mode)){
                    RpcFiltersContainer.getServerFilter().get(mode).add((RpcServerFilter) bean);
                }else{
                    List<RpcServerFilter> map = new ArrayList<>();
                    map.add((RpcServerFilter)bean);
                    RpcFiltersContainer.getServerFilter().put(mode,map);
                }
            });

        }
        if(log.isInfoEnabled()){
            log.info("Channel "+this.channelType()+" Client filters:"+ SerializerFactory.getSerializer(SerializeType.Json).serializeToString(RpcFiltersContainer.getClientFilter()));
            log.info("Channel "+this.channelType()+" Server filters:"+ SerializerFactory.getSerializer(SerializeType.Json).serializeToString(RpcFiltersContainer.getServerFilter()));
        }

    }

//    /**
//     * CAF RPC Client Filter
//     * current request filters
//     * the global filters is RpcFilterContainer
//     */
//    @Setter
//    @Getter
//    protected List<RpcClientFilter> clientFilters;
//    /**
//     * CAF RPC Server Filter
//     * current request filters
//     * the global filters is RpcFilterContainer
//     */
//    @Setter
//    @Getter
//    protected List<RpcServerFilter> serverFilters;

    /**
     * target host & port to initialize client
     */
    @Getter
    protected String host;
    @Getter
    protected int port;

    protected String filterType;

//    /**
//     * RPC Client
//     * different channel has own different implement
//     */
//    @Getter
//    private Object client;
//    /**
//     * RPC Server
//     * different channel has own different implement
//     */
//    @Getter
//    private Object server;

    //public static void serviceRegister();

    /**
     * subClass implement this method to identify the custom channel type
     * this type must Globally Unique
     * @return channelType
     */
    public abstract String channelType();

    /**
     * subClass implement this method to build custom channel
     * the channel must create client,server,client filter,server filter properties
     * @param host host
     * @param port port
     * @return RpcChannel
     */
    public abstract RpcChannel buildChannel(String host, int port);

    /**
     * subClass implement this method to build custom channel
     * this method allow you to define a custom filter channel
     * @param host host
     * @param port port
     * @param filterType filter type{@link #channelType}
     * @return RpcChannel
     */
    public abstract RpcChannel buildChannel(String host, int port,String filterType);

    public abstract RpcChannel buildChannel(URL url, String filterType);

    /**
     * subClass to implement this method to add headers info to client
     */
    public abstract void addHeaders(Map<String,String> context);


    /**
     * start a channel server
     */
    public abstract void startServer(Map<String,Object> args);

    /**
     * stop a channel server
     */
    protected abstract void stopServer();


    /**
     * start a channel client
     */
    public abstract void startClient();

    /**
     * stop a channel client
     */
    public abstract void stopClient();
}
